#include "common.h"


int UpstreamDefaultHandler(Upstream *up)
{
    return HTTP_AGAIN;
}
Upstream * CreateUpsteam(Task *task, int protocal, char *ip, int port)
{
    Upstream *upstream;
    int s;
    Conn *conn;

    conn = task->conn;
    
    upstream = DS_Pcalloc(conn->pool,sizeof(Upstream));
    if (upstream == NULL) {
        Log_Write(g_server->log, "Upstream: No Memory For a new upstream");
        return NULL;
    }
    memcpy(upstream->host, ip , strlen(ip));
    upstream->port = port;
    
    s = NET_CreateSocket(AF_INET);
    if (s == NET_ERR) {
        Log_Write(g_server->log, "Upstream: Create Socket Failed");
        return NULL;
    }
    if (NET_NonBlock(s) == NET_ERR) {
        Log_Write(g_server->log, "Upstream: Set Nonblock Failed");
        return NULL;
    }
    
    upstream->peer = CONN_Get_Connection(s);
    if (upstream->peer == NULL) {
        Log_Write(g_server->log, "Upstream: No Conn");
        return NULL;
    }
    upstream->peer->pool = conn->pool;
    upstream->peer->data = upstream;
    upstream->parent = task;
    if (EVENT_Handle_Add_Connection(upstream->peer) != EVENT_OK) {
        CONN_Close_Connection(upstream->peer);
        upstream->err = 1;
        Log_Write(g_server->log, "Upstream: Add Epoll Conn Failed");
        return NULL;
    }

    if (protocal == UPSTREAM_PROTOCAL_MC) {
        upstream->handler = MC_ParseResponse;
    } else if (protocal == UPSTREAM_PROTOCAL_RAW){
        upstream->handler = UpstreamDefaultHandler;
    } else {
        Log_Write(g_server->log, "Upstream: Now Only Support MC protocal");
        return NULL;
    }

    return upstream;
    
}

int AddUpstreamRequestBuf(Upstream *upstream, char *buffer, size_t len)
{
    Conn *c;
    Chain *chain, *temp;
    Buf *buf;

    c = upstream->peer;
    chain = DS_Pcalloc(c->pool, sizeof(Chain));
    if (chain == NULL) {
        return HTTP_ERR;
    }

    buf = DS_CreateTempBuf(c->pool, len);
    memcpy(buf->last, buffer, len);
    buf->last += len;
    chain->buf = buf;
    chain->next = NULL;
    
    if (upstream->request_buf == NULL) {
        upstream->request_buf = chain;
        return HTTP_OK;
    }

    temp = upstream->request_buf;
    while (temp->next) {
        temp = temp->next;
    }
    temp->next = chain;  

    return HTTP_OK;
}

Task * CreateTaskTable(Conn *c, async_event_done_pt handler)
{
    Task *task;

    task = DS_Pcalloc(c->pool ,sizeof(Task));
    if (task == NULL) {
        Log_Write(g_server->log, "Upstream: No Memory For Task");
        return NULL;
    }
    task->handler = handler;
    task->upstreams = DS_ArrayCreate(c->pool,5,sizeof(Upstream*));
    if (task->upstreams == NULL) {
        Log_Write(g_server->log, "Upstream: No Memory For Task->upstreams");
    }
    task->conn = c;
    c->task = task;

    return task;
}

int AddTask(Upstream *upstream, Task *task)
{
    Upstream **up;
    
    up = DS_ArrayPush(task->upstreams);
    if (up == NULL) {
        return HTTP_ERR;
    }
    *up = upstream;
    return HTTP_OK;
}

int ConnectPeer(Upstream *uptream)
{
    struct sockaddr_in sa;
    int rc;
    Conn *c = uptream->peer;

    sa.sin_family = AF_INET;
    sa.sin_port = htons(uptream->port);
    sa.sin_addr.s_addr = inet_addr(uptream->host);

    rc = connect(c->fd, (struct sockaddr *)&sa , sizeof(sa));
    if (rc == -1) {
        if (errno != EINPROGRESS) {
            return NET_ERR;
        } else {
            return NET_AGAIN;
        }
    }

    c->write->ready = 1;
    return NET_OK;
    
}

int Upstream_TestConnect(Upstream *upstream)
{
    int err = 0;
    socklen_t   len = sizeof(int);

        /*
         * BSDs and Linux return 0 and set a pending error in err
         * Solaris returns -1 and sets errno
         */

    if (getsockopt(upstream->peer->fd, SOL_SOCKET, SO_ERROR, (void *) &err, &len) == -1)
    {
        err = errno;
    }

    if (err) {
        return -1;
    }

    return 0;
}


void UpstreamDummyHandler(Event *ev)
{

}


int UpstreamReadChain(Upstream *upstream)
{
    Conn *conn = upstream->peer;
    Chain *temp, **ll;
    Buf *buf;
    Chain *chain, *tt;
    int rc;
    size_t len;
    int newchain = 1;  /* indicate whether use the last buf or alloc a new one  */

    ll = &upstream->response_buf;
    tt = upstream->response_buf;
    for (chain = upstream->response_buf; chain; chain = chain->next) {
        ll = &chain->next;
        tt = chain;
    }
    
    while(1) 
    {
        /* Try to make full use of a buf space */
        if (tt && tt->buf->end - tt->buf->last > 0) {
            newchain = 0;
            buf = tt->buf;
        } else {
            /* Here, there will be one more buf when done, and buf->last == buf->pos, Be careful! */
            newchain = 1;
            temp = DS_AllocChainLink(conn->pool);
            buf = DS_CreateTempBuf(conn->pool, 1024);
            
	        if (buf == NULL) {
                Log_Write(g_server->log,"Upstream: No Memory");
		        return NET_ERR;
            } 
            temp->buf = buf;
            temp->next = NULL;

            *ll = temp;
            tt = temp;
            ll = &temp->next;
        }
        
        len = buf->end - buf->last;
        rc = read(conn->fd, buf->last, len);
        if (rc > 0) {
            buf->last += rc;
            upstream->rcvlen += rc;
        } else if(rc == 0){ 
            Log_Write(g_server->log,"Process: %d  SocketID: %d  Recv %d bytes", g_Pid, conn->fd, rc);
            return NET_CLOSE;
        } else {
            if (errno == EAGAIN) {
                return NET_AGAIN;
            } else if (errno == EINTR){
                continue;
            } else {
                return NET_ERR;
            }
        }
    }

    *ll = NULL;
    
    return NET_OK;
}

int UpstreamWriteChain(Upstream *upstream)
{
    Conn *conn = upstream->peer;
    Chain *chain = upstream->request_buf;
 

    off_t   size;
    Chain   *cl, *ln, **ll;

    size = 0;
    ll = &upstream->request_buf;

    for (cl = upstream->request_buf; cl; cl = cl->next) {
        ll = &cl->next;
        size += DS_BUF_SIZE(cl->buf);
    }

    *ll = NULL;

    chain = SendChain(conn, upstream->request_buf);

    if (chain == ((Chain*)-1)) {
        return NET_ERR;
    }

    for (cl = upstream->request_buf; cl && cl != chain; /* void */) {
        ln = cl;
        cl = cl->next;
        DS_FreeChain(conn->pool, ln);
    }

    upstream->request_buf = chain;

    if (chain) {
        return NET_AGAIN;
    }
    
    return NET_OK;
    
}

void UpstreamSendRequest(Event *ev)
{
    Conn *conn;
    Upstream *upstream;
    int rc;
    
    conn = ev->data;
    upstream = conn->data;

    if (Upstream_TestConnect(upstream) != 0) {
        Log_Write(g_server->log, "%d Connect %s : %d failed ", upstream->peer->fd, upstream->host, upstream->port);
        CONN_Close_Connection(upstream->peer);
        upstream->err = 1;
        FinalizeConn(upstream->parent->conn);
        return;
    }
    
    rc = UpstreamWriteChain(upstream);
    if (rc == NET_OK) {
        upstream->peer->write->handler = UpstreamDummyHandler;
    } else if (rc == NET_AGAIN) {
        Log_Write(g_server->log, "Upstream: Send Again");
        EVENT_Handle_Write_Event(ev,EVENT_CLEAR_EVENT);
    } else {
        Log_Write(g_server->log, "Upstream: Send Err");
        CONN_Close_Connection(upstream->peer);
        upstream->err = 1;
        FinalizeConn(upstream->parent->conn);
        return;
    }
}

int AppendUpstreamResponseBuf(Upstream *upstream, char *buffer, int len)
{
    Chain *chain, *temp;
    Buf *buf;
    Conn *c = upstream->peer;
    
    chain = DS_Pcalloc(c->pool, sizeof(Chain));
    if (chain == NULL) {
        return DS_ERROR;
    }

    buf = DS_CreateTempBuf(c->pool, len);
    memcpy(buf->last, buffer, len);
    buf->last += len;
    chain->buf = buf;
    chain->next = NULL;
    
    if (upstream->response_buf == NULL) {
        upstream->response_buf = chain;
        return DS_OK;
    }

    temp = upstream->response_buf;
    while (temp->next) {
        temp = temp->next;
    }
    temp->next = chain;
    
}

int CheckTaskDone(Task *task)
{
    int i;
    Upstream *up;
    int rc;
    
    ds_array_s *array = task->upstreams;
    
    for (i = 0; i < array->nelts; i++) {
        up = ((Upstream**)array->elts)[i];
        if (!up->eof && !up->err && !up->complete) {
            return -1;
        }
    }

    return 0;
}

void UpstreamProcessRequest(Event *ev)
{
    Conn *conn;
    Upstream *upstream;
    int rc, rv;

    conn = ev->data;
    upstream = conn->data;

    if (Upstream_TestConnect(upstream) != 0) {
        Log_Write(g_server->log, "%d Connect %s : %d failed ", upstream->peer->fd, upstream->host, upstream->port);
        CONN_Close_Connection(upstream->peer);
        upstream->err = 1;
        FinalizeConn(upstream->parent->conn);
        return;
    }
    
    rc = UpstreamReadChain(upstream);

    if (rc == NET_OK || rc == NET_AGAIN) {
        rv = upstream->handler(upstream);
        if (rv == HTTP_OK) {
            upstream->complete = 1;
            //Log_Write(g_server->log, "Upsream: %d Parse Response Success", upstream->peer->fd);
            CONN_Close_Connection(upstream->peer);
        } else if (rv == HTTP_AGAIN) {
            Log_Write(g_server->log, "Upsream: %d Parse Response Again", upstream->peer->fd);
            EVENT_Handle_Read_Event(ev, EVENT_CLEAR_EVENT);
        } else {
            Log_Write(g_server->log, "Upsream: %d Parse Response Failed", upstream->peer->fd);
            CONN_Close_Connection(upstream->peer);
            upstream->err = 1;
        }   
    } else if (rc == NET_CLOSE) {
    	rv = upstream->handler(upstream);
    	if (rv == HTTP_OK) {
            upstream->complete = 1;
            Log_Write(g_server->log, "Upsream: %d Parse Response Success", upstream->peer->fd);
            CONN_Close_Connection(upstream->peer);
        } else {
            Log_Write(g_server->log, "Upsream: %d Peer Close", upstream->peer->fd);
            CONN_Close_Connection(upstream->peer);
            upstream->eof = 1;
	    }
    } else {
        Log_Write(g_server->log, "Upsream: Read Failed");
        CONN_Close_Connection(upstream->peer);
        upstream->err = 1;
        FinalizeConn(upstream->parent->conn);
        return;
    }

    if (CheckTaskDone(upstream->parent) == 0) {
        rc = upstream->parent->handler(upstream->parent);
    }
    
}

int RunTaskEngine(Task *task)
{
    int i;
    Upstream *up;
    int rc;
    
    ds_array_s *array = task->upstreams;
    
    for (i = 0; i < array->nelts; i++) {
        up = ((Upstream**)array->elts)[i];
        rc = ConnectPeer(up);
        if (rc == NET_ERR) {
            CONN_Close_Connection(up->peer);
            up->err = 1;
            Log_Write(g_server->log, "Connect Peer Err %d", errno);
            FinalizeConn(task->conn);
            break;
        } 
        up->peer->write->handler = UpstreamSendRequest;
        up->peer->read->handler = UpstreamProcessRequest;
        if (rc == NET_AGAIN) {
            continue;
        }
        if (rc == NET_OK) {
            Log_Write(g_server->log, "Connect Success Now");
            UpstreamSendRequest(up->peer->write);
        }
    }
}

